#include "mqtt_task.h"
#include <etimer.h>
#include <process.h>
#include <string.h>
#include "MQTTPacket.h"
#include "stdlib.h"

#include "convert.h"
#include "device_manage.h"
#include "fun_flash.h"
#include "gprs_task.h"
#include "gsm.h"
#include "json.h"
#include "memb.h"
#include "net_api.h"
#include "time.h"
#include "uart.h"
#include "wifi.h"
#include "wifi_task.h"
#include "disp_task.h"
#include "lora_task.h"



static uint8_t i = 0;
static sWifiCmd wificmd;
static sGprsCmd gprscmd;

static int buflen = 256;
static unsigned char buf[256];

static unsigned long last_ping_time;
static int len = 0;
static int tmp;

static char res = 0;
static struct process *pro;

static MQTTPacket_connectData mqtt_data;
static unsigned short msgid_t = 0;
static unsigned short msgid_r = 0;
static unsigned char dup;
static int qos;
static unsigned char retained;
static int payloadlen_in;
static unsigned char *payload_in;
static int rc;
static MQTTString Topic;

static my_cJSON *root = NULL;
static my_cJSON *jsontime = NULL;
static my_cJSON *jsoncmd = NULL;
static my_cJSON *jsonnum = NULL;
static my_cJSON *jsonmac = NULL;

extern FIFO_Type mqtt_fifo;
static unsigned char databuf[256];

extern struct process LORA_Process;

MEMB(m_PackageBLE, PackageBLE, 16);
MEMB(m_PackageST, PackageST, 16);

void menb_pack_init(void) {
  memb_init(&m_PackageBLE);
  memb_init(&m_PackageST);
}

static int getdata(unsigned char *buf, int count) {
  return uart_receive(buf, count);
  ;
}

void send_callnum(PackageST st_package) {
  sMqttData MqttData;

  if (cfg_data.Val.AddFlag == HaveAdd) {
    PackageST *package = (PackageST *)memb_alloc(&m_PackageST);
    package->mac = st_package.mac;
    package->power = st_package.power;
	MqttData.eve = topic_eve_data;
    MqttData.cmd = Call_Exte;
    MqttData.dup = 0;
    MqttData.data = package;
    FIFO_AddOne(&mqtt_fifo, &MqttData);
	printf("PackageST memb_numfree:%d\r\n",memb_numfree(&m_PackageST));
  }
}

void send_callnum_location(PackageBLE ble_package) {
  sMqttData MqttData;
  if (cfg_data.Val.AddFlag == HaveAdd) {
    PackageBLE *package = (PackageBLE *)memb_alloc(&m_PackageBLE);
    package->mac = ble_package.mac;
    package->power = ble_package.power;
    memcpy(package->location.mac, ble_package.location.mac, 6);

	MqttData.eve = topic_eve_data;
    MqttData.cmd = Call_BLE_Exte;
    MqttData.dup = 0;
    MqttData.data = package;
    FIFO_AddOne(&mqtt_fifo, &MqttData);
	printf("PackageBLE memb_numfree:%d\r\n",memb_numfree(&m_PackageBLE));
  }
}

void send_register(PackageST st_package) {
  sMqttData MqttData;
  if (cfg_data.Val.AddFlag == HaveAdd) {
    PackageST *package = (PackageST *)memb_alloc(&m_PackageST);
    package->mac = st_package.mac;
    package->power = st_package.power;
	  
	MqttData.eve = topic_eve_data;
    MqttData.cmd = Reg_Exte;
    MqttData.dup = 0;
    MqttData.data = package;
    FIFO_AddOne(&mqtt_fifo, &MqttData);
	
	printf("PackageST memb_numfree:%d\r\n",memb_numfree(&m_PackageST));
  }
}

void send_gettime() {
  static sMqttData MqttData;
  if (cfg_data.Val.AddFlag == HaveAdd) {
	  
	MqttData.eve = topic_eve_data;
    MqttData.cmd = Req_Time;
    MqttData.dup = 0;
    FIFO_AddOne(&mqtt_fifo, &MqttData);
  }
}

void send_connect() {
  static sMqttData MqttData;

	MqttData.eve = topic_eve_conn;
    MqttData.cmd = 0;
    MqttData.dup = 0;
    FIFO_AddOne(&mqtt_fifo, &MqttData);

}

void send_addhost_ack() {
  static sMqttData MqttData;
  // if(cfg_data.Val.AddFlag == NoneAdd){
  MqttData.eve = topic_eve_data;
  MqttData.cmd = Add_Host_Ack;
  MqttData.dup = 0;
  FIFO_AddOne(&mqtt_fifo, &MqttData);
  //}
}

void send_delhost_ack() {
  static sMqttData MqttData;
  //	if(cfg_data.Val.AddFlag == HaveAdd){

  MqttData.eve = topic_eve_data;
  MqttData.cmd = Del_Host_Ack;
  MqttData.dup = 0;
  FIFO_AddOne(&mqtt_fifo, &MqttData);
  //	}
}

void send_addexte_ack(iSMacSn mac) {
  static sMqttData MqttData;
  static char str[25] = {0};
  if (cfg_data.Val.AddFlag == HaveAdd) {
    memset(str, 0, 25);
    sprintf(str, "%.8X%.8X%.8X", mac.temp0, mac.temp1, mac.temp2);
    str[24] = 0;

	MqttData.eve = topic_eve_data;
    MqttData.cmd = Add_Exte_Ack;
    MqttData.dup = 0;
    MqttData.data = str;

    FIFO_AddOne(&mqtt_fifo, &MqttData);
  }
}

void send_modexte_ack(iSMacSn mac) {
  static sMqttData MqttData;
  static char str[25] = {0};
  if (cfg_data.Val.AddFlag == HaveAdd) {
    memset(str, 0, 25);
    sprintf(str, "%.8X%.8X%.8X", mac.temp0, mac.temp1, mac.temp2);
    str[24] = 0;

	MqttData.eve = topic_eve_data;
    MqttData.cmd = Mod_Exte_Ack;
    MqttData.dup = 0;
    MqttData.data = str;

    FIFO_AddOne(&mqtt_fifo, &MqttData);
  }
}

void send_delexte_ack(iSMacSn mac) {
  static sMqttData MqttData;
  static char str[25] = {0};
  if (cfg_data.Val.AddFlag == HaveAdd) {
    memset(str, 0, 25);
    sprintf(str, "%.8X%.8X%.8X", mac.temp0, mac.temp1, mac.temp2);
    str[24] = 0;

	MqttData.eve = topic_eve_data;
    MqttData.cmd = Del_Exte_Ack;
    MqttData.dup = 0;
    MqttData.data = str;

    FIFO_AddOne(&mqtt_fifo, &MqttData);
  }
}

PROCESS(MqttCFG_Process, "MqttCFG");
PROCESS_THREAD(MqttCFG_Process, ev, data) {
  static struct etimer et;

  PROCESS_BEGIN();
  etimer_set(&et, 3000);
  PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&et));
  write_cfg(&cfg_data);
  PROCESS_END();
}

void add_hostdev() {
  if (cfg_data.Val.AddFlag == HaveAdd) {
    send_addhost_ack();
  } else {  //后台添加主机，启动lora任务
    //		process_start(&LORA_Process, NULL);
    send_addhost_ack();
    cfg_data.Val.AddFlag = HaveAdd;
    //		write_cfg(&cfg_data);
    process_start(&MqttCFG_Process, NULL);
  }
}

void del_hostdev() {
  if (cfg_data.Val.AddFlag == HaveAdd) {  //后台删除主机，退出lora任务
    send_delhost_ack();
    cfg_data.Val.AddFlag = NoneAdd;
    //		write_cfg(&cfg_data);
    process_start(&MqttCFG_Process, NULL);
    del_allflashnum();  //删除所有分机
                        //		process_exit(&LORA_Process);
  } else {
    send_delhost_ack();
  }
}

void add_exte(char *mach, uint16_t num) {
  static char buf[9];
  static iSMacSn mac;

  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp0 = htoi(buf);
  mach = mach + 8;
  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp1 = htoi(buf);
  mach = mach + 8;
  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp2 = htoi(buf);

  if (!check_flashnum(mac)) {  // flash中不存在该mac，就添加
    add_flashnum(mac, num);
  }
  send_addexte_ack(mac);
}

void mod_exte(char *mach, uint16_t num) {
  static char buf[9];
  static iSMacSn mac;

  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp0 = htoi(buf);
  mach = mach + 8;
  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp1 = htoi(buf);
  mach = mach + 8;
  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp2 = htoi(buf);

  if (!check_flashnum(mac)) {  // flash中不存在该mac，就添加
    add_flashnum(mac, num);
  } else {
    mod_flashnum(num);
  }
  send_modexte_ack(mac);
}

void del_exte(char *mach) {
  static char buf[9];
  static iSMacSn mac;

  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp0 = htoi(buf);
  mach = mach + 8;
  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp1 = htoi(buf);
  mach = mach + 8;
  memset(buf, '\0', 9);
  memcpy(buf, mach, 8);
  mac.temp2 = htoi(buf);

  if (check_flashnum(mac)) {  //检测是否有该mac的分机
    del_flashnum();
  }
  send_delexte_ack(mac);
}

void upd_time(uint32_t time) { TimeDate_Setsec(time + 28800); }

void clear_disp_num(uint16_t num)
{
	static sLEDData set_data;
	remove_dispnum(num);
	remove_callnum(num);
    set_data.cmd = disp_num;
    set_data.count = 1;
	process_post(&Disp_Process, PROCESS_EVENT_MSG, (process_data_t)&set_data);
}

void analysis_data(uint8_t *data, uint16_t len) {
  static char *s = NULL;
  static uint8_t *prt = NULL;
  uint8_t cmd = 0;
  uint16_t num = 0;
  char *mach = NULL;

  prt = data;
  s = strstr((const char *)prt, "{");
  root = my_cJSON_Parse((const char *)s);

  if (root == NULL) {
    return;
  }
  jsoncmd = my_cJSON_GetObjectItem(root, "cmd");
  cmd = jsoncmd->valueint;

  if (cmd == Add_Host) {
	printf("mqtt sub:cmd->Add_Host\r\n");
    add_hostdev();
  } else if (cmd == Del_Host) {
	printf("mqtt sub:cmd->Del_Host\r\n");
    del_hostdev();
  }

  if (cfg_data.Val.AddFlag == HaveAdd) {
    if (cmd == Add_Exte) {
	  printf("mqtt sub:cmd->Add_Exte\r\n");
      jsonmac = my_cJSON_GetObjectItem(root, "mac");
      jsonnum = my_cJSON_GetObjectItem(root, "num");
      num = jsonnum->valueint;
      mach = jsonmac->valuestring;
      add_exte(mach, num);
    } else if (cmd == Mod_Exte) {
	  printf("mqtt sub:cmd->Mod_Exte\r\n");
      jsonmac = my_cJSON_GetObjectItem(root, "mac");
      jsonnum = my_cJSON_GetObjectItem(root, "num");
      num = jsonnum->valueint;
      mach = jsonmac->valuestring;
      mod_exte(mach, num);
    } else if (cmd == Del_Exte) {
	  printf("mqtt sub:cmd->Del_Exte\r\n");
      jsonmac = my_cJSON_GetObjectItem(root, "mac");
      if (jsonmac != NULL) {
        mach = jsonmac->valuestring;
        del_exte(mach);
      }
    } else if (cmd == Upd_Time) {
	  printf("mqtt sub:cmd->Upd_Time\r\n");
      jsontime = my_cJSON_GetObjectItem(root, "time");
      upd_time(jsontime->valueint);
    } else if (cmd == ClrDisp_Exte) {
	  printf("mqtt sub:cmd->ClrDisp_Exte\r\n");
      jsonnum = my_cJSON_GetObjectItem(root, "num");
      clear_disp_num(jsonnum->valueint);
    }
  }

  my_cJSON_Delete(root);
}

static void connect_Serialize() {
  static char topic[64] = {0};
  cfg_data.Val.Device_id[16] = '\0';
  mqtt_data.clientID.lenstring.data = (char *)cfg_data.Val.Device_id;
  mqtt_data.clientID.lenstring.len = 16;
  mqtt_data.cleansession = 1;
  mqtt_data.keepAliveInterval = 20;
  mqtt_data.willFlag = 1;
  mqtt_data.will.message.lenstring.data = (char *)cfg_data.Val.Device_id;
  mqtt_data.will.message.lenstring.len = 16;
  mqtt_data.will.qos = 1;
  mqtt_data.will.retained = 0;
  mqtt_data.username.cstring = "jzl";
  mqtt_data.password.cstring = "jzl";
  memset(topic, '\0', 64);
  strcat(topic, Pro_topic);  			//一级主题 产品
  strcat(topic, "/");
  strcat(topic, DevType_topic);			//二级主题 设备类型
  strcat(topic, "/");
  strcat(topic, DevVers_topic);			//三级主题 设备版本
  strcat(topic, "/");
  strcat(topic, ObjSys_topic);			//四级主题 发往对象
  strcat(topic, "/");
  strcat(topic, EveLwt_topic);			//五级主题 离线事件
  mqtt_data.will.topicName.lenstring.data = topic;
  mqtt_data.will.topicName.lenstring.len = strlen((const char*) topic);
  mqtt_data.MQTTVersion = 4;
  len = MQTTSerialize_connect(buf, buflen, &mqtt_data);
}

static void sub_Serialize() {
  MQTTString topicString = MQTTString_initializer;
  int req_qos = SUB_QOS;
  int top_len = 0;
  static char topic[64] = {0};

  //订阅 jzl/产品ID
  memset(topic, '\0', 64);
  strcat(topic, Pro_topic);  						//一级主题 产品
  strcat(topic, "/");
  strcat(topic, DevType_topic);						//二级主题 设备类型
  strcat(topic, "/");
  strcat(topic, DevVers_topic);						//三级主题 设备版本
  strcat(topic, "/");
  strcat(topic, (char *)cfg_data.Val.Device_id); 	//四级主题 发往对象
  strcat(topic, "/");
  strcat(topic, EveData_topic);						//五级主题 数据事件

  top_len = strlen(topic);
  topicString.lenstring.data = topic;
  topicString.lenstring.len = top_len;
  ++msgid_t;
  len = MQTTSerialize_subscribe(buf, buflen, 0, msgid_t, 1, &topicString,
                                &req_qos);
}

uint8_t data_len = 0;
static uint8_t data_Serialize(sMqttData MqttData) {
	
  root = my_cJSON_CreateObject();

  if(MqttData.eve == topic_eve_data){
	  switch (MqttData.cmd) {
		case Add_Exte_Ack: {
		  my_cJSON_AddNumberToObject(root, "cmd", MqttData.cmd);
		  my_cJSON_AddStringToObject(root, "mac", (const char *)MqttData.data);
		  break;
		}
		case Mod_Exte_Ack: {
		  my_cJSON_AddNumberToObject(root, "cmd", MqttData.cmd);
		  my_cJSON_AddStringToObject(root, "mac", (const char *)MqttData.data);
		  break;
		}
		case Del_Exte_Ack: {
		  my_cJSON_AddNumberToObject(root, "cmd", MqttData.cmd);
		  my_cJSON_AddStringToObject(root, "mac", (const char *)MqttData.data);
		  break;
		}
		case Req_Time:
		case Add_Host_Ack:
		case Del_Host_Ack:{
		  my_cJSON_AddNumberToObject(root, "cmd", MqttData.cmd);
		  break;
		}
		case Reg_Exte:
		case Call_Exte: {
		  char str[25] = {0};
		  PackageST *package = (PackageST *)MqttData.data;
		  sprintf(str, "%.8X%.8X%.8X", package->mac.temp0, package->mac.temp1,
				  package->mac.temp2);
		  my_cJSON_AddNumberToObject(root, "cmd", MqttData.cmd);
		  my_cJSON_AddStringToObject(root, "mac", (const char *)str);
		  my_cJSON_AddNumberToObject(root, "pow", package->power);
		  // memb_free(&m_PackageST, package);
		  break;
		}
		case Call_BLE_Exte: {
		  char str[25] = {0};
		  PackageBLE *package = (PackageBLE *)MqttData.data;
		  sprintf(str, "%.8X%.8X%.8X", package->mac.temp0, package->mac.temp1,
				  package->mac.temp2);
		  my_cJSON_AddNumberToObject(root, "cmd", MqttData.cmd);
		  my_cJSON_AddStringToObject(root, "mac", (const char *)str);
		  my_cJSON_AddNumberToObject(root, "pow", package->power);
		  memset(str, 0, 25);
		  sprintf(str, "%.2X%.2X%.2X%.2X%.2X%.2X", package->location.mac[0],
				  package->location.mac[1], package->location.mac[2],
				  package->location.mac[3], package->location.mac[4],
				  package->location.mac[5]);
		  my_cJSON_AddStringToObject(root, "bmac", (const char *)str);
		  // memb_free(&m_PackageBLE, package);
		  break;
		}
	  }
  }else{
	 my_cJSON_AddStringToObject(root, "did", (const char *)cfg_data.Val.Device_id);
  }
  char *s = my_cJSON_PrintUnformatted(root);
  if (s == 0) {
    my_cJSON_Delete(root);
    HAL_NVIC_SystemReset();
    return 0;
  }
  data_len = strlen((const char *)s);
  memset(databuf, 0, 256);
  memcpy((char *)databuf, s, data_len);

  free(s);
  my_cJSON_Delete(root);

  return data_len;
}

void free_mqttdata(sMqttData MqttData) {
  if(MqttData.eve == topic_eve_conn){
	return;
  }
  switch (MqttData.cmd) {
    case Add_Host_Ack:
    case Del_Host_Ack:
    case Add_Exte_Ack:
    case Mod_Exte_Ack:
    case Del_Exte_Ack: {
      break;
    }
    case Reg_Exte:
    case Call_Exte: {
      PackageST *package = (PackageST *)MqttData.data;
      memb_free(&m_PackageST, package);
      break;
    }
    case Call_BLE_Exte: {
      PackageBLE *package = (PackageBLE *)MqttData.data;
      memb_free(&m_PackageBLE, package);
      break;
    }
    default: { break; }
  }
}

static uint8_t pub_Serialize(uint8_t topic_event, uint8_t *msg, uint8_t msglen, unsigned char dup,
                             int qos, unsigned short *msgid) {
  static char topic[32] = {0};
  static unsigned short id = 0;
  MQTTString topicString = MQTTString_initializer;
  int top_len = 0;
  uint8_t len;

  //订阅 jzl/产品ID
  memset(topic, '\0', 64);
  strcat(topic, Pro_topic);  						//一级主题 产品
  strcat(topic, "/");
  strcat(topic, DevType_topic);						//二级主题 设备类型
  strcat(topic, "/");
  strcat(topic, DevVers_topic);						//三级主题 设备版本
  strcat(topic, "/");
  strcat(topic, ObjSys_topic); 						//四级主题 发往对象
  strcat(topic, "/");
  if(topic_event == topic_eve_conn){
	strcat(topic, EveConn_topic);						//五级主题 数据事件
  }else{
	strcat(topic, EveData_topic);						//五级主题 数据事件
  }
  
  top_len = strlen(topic);
  topicString.lenstring.data = topic;
  topicString.lenstring.len = top_len;

  if (dup == 1) {
    id = *msgid;
  } else {
    id = ++msgid_t;
    *msgid = id;
  }
  printf("mqtt send:id->%d\r\n",id);
  len = MQTTSerialize_publish(buf, buflen, dup, qos, 0, id, topicString, msg,
                              msglen);

  return len;
}

PROCESS(MqttInit_Process, "MqttInit");
PROCESS_THREAD(MqttInit_Process, ev, data) {
  static uint8_t i = 0;
  static struct etimer et;
  PROCESS_BEGIN();
  
  printf("MqttInit_Process start  \r\n");
  if (netdevice_type == net_device_wifi) {
    process_exit(&Wifi_GetIPD_Process);
  } else {
    process_exit(&Gprs_GetIPD_Process);
  }
  //连接Mqtt服务器
  process_start(&MqttConnect_Process, (process_data_t)&MqttInit_Process);
  do {
    PROCESS_WAIT_EVENT();
  } while (ev != PROCESS_EVENT_CONTINUE);
  if (*(uint8_t *)data == 1) {
  } else {
    if (netdevice_type == net_device_wifi) {
      etimer_set(&et, 1000);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      uart_send("+", 1);
      etimer_set(&et, 10);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      uart_send("+", 1);
      etimer_set(&et, 10);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      uart_send("+", 1);
      etimer_set(&et, 100);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      i = 1;
      set_wificmdval(&wificmd, AT_OUTTRANS, AT_OUTTRANS_RES, &MqttInit_Process);
      wifisend_cmd(WifiCMD_Process, wificmd, i);
      process_start(&WifiInit_Process, NULL);
    } else {
      etimer_set(&et, 3100);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      Gsm_CIPTMODE_Off();
      etimer_set(&et, 1000);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      i = 1;
      set_gprscmdval(&gprscmd, GSM_CIPCLOSE, GSM_CIPCLOSE_RES, GSM_ERROE_RES,
                     &MqttInit_Process);
      gprssend_cmd(GprsCMD_Process, gprscmd, i);
      process_start(&GprsInit_Process, NULL);
    }
    process_exit(&MqttInit_Process);
    return PT_ENDED;
  }

  //订阅主题

  process_start(&MqttSub_Process, (process_data_t)&MqttInit_Process);
  do {
    PROCESS_WAIT_EVENT();
  } while (ev != PROCESS_EVENT_CONTINUE);
  if (*(uint8_t *)data == 1) {
  } else {
    if (netdevice_type == net_device_wifi) {
      etimer_set(&et, 1000);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      uart_send("+", 1);
      etimer_set(&et, 10);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      uart_send("+", 1);
      etimer_set(&et, 10);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      uart_send("+", 1);
      etimer_set(&et, 100);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      i = 1;
      set_wificmdval(&wificmd, AT_OUTTRANS, AT_OUTTRANS_RES, &MqttInit_Process);
      wifisend_cmd(WifiCMD_Process, wificmd, i);
      process_start(&WifiInit_Process, NULL);
    } else {
      etimer_set(&et, 3100);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      Gsm_CIPTMODE_Off();
      etimer_set(&et, 1000);
      PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      i = 1;
      set_gprscmdval(&gprscmd, GSM_CIPCLOSE, GSM_CIPCLOSE_RES, GSM_ERROE_RES,
                     &MqttInit_Process);
      gprssend_cmd(GprsCMD_Process, gprscmd, i);
      process_start(&GprsInit_Process, NULL);
    }
    process_exit(&MqttInit_Process);
    return PT_ENDED;
  }
  
  send_connect();

  // MQTT 循环检测任务
  process_start(&MqttLoop_Process, NULL);

  PROCESS_END();
}

PROCESS(MqttConnect_Process, "MqttConnect");
PROCESS_THREAD(MqttConnect_Process, ev, data) {
  static struct etimer et;
  static unsigned long tmp_time;
  static uint8_t datalen = 0;
  static uint8_t pre_datalen = 0;
  PROCESS_BEGIN();

  res = 0;
  netstatus = net_doing;

  if (ev == PROCESS_EVENT_INIT) {
    pro = (struct process *)data;
  }

  if (netdevice_type == net_device_wifi) {
    i = 1;
    set_wificmdval(&wificmd, AT_CIPMQTT, AT_CIPMQTT_RES, &MqttConnect_Process);
    wifisend_cmd(WifiCMD_Process, wificmd, i);
    etimer_set(&et, 200);
    PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
    wifisend_cmd(WifiCMD_Process, wificmd, i);
    i = 1;
    set_wificmdval(&wificmd, AT_ENTERTRANS, AT_ENTERTRANS_RES,
                   &MqttConnect_Process);
    wifisend_cmd(WifiCMD_Process, wificmd, i);
    i = 1;
    set_wificmdval(&wificmd, AT_TRANSSEND, AT_TRANSSEND_RES,
                   &MqttConnect_Process);
    wifisend_cmd(WifiCMD_Process, wificmd, i);
    etimer_set(&et, 20);
    PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
  } else {
    i = 1;
    set_gprscmdval(&gprscmd, GSM_CIPMQTT, GSM_CIPMQTT_RES, GSM_ERROE_RES,
                   &MqttConnect_Process);
    gprssend_cmd(GprsCMD_Process, gprscmd, i);
    etimer_set(&et, 200);
    PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
    Gsm_CIPTMODE_On();
    etimer_set(&et, 300);
    PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
  }

  uart_clear();

  connect_Serialize();  //连接数据序列化
  uart_send((uint8_t *)buf, len);
  memset(buf, 0, buflen);

  pre_datalen = datalen = uart_getlen();
  tmp_time = HAL_GetTick();
  while (1) {
    etimer_set(&et, 10);
    PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
    datalen = uart_getlen();
    if ((datalen == pre_datalen && datalen > 0) ||
        (HAL_GetTick() - tmp_time > 5000)) {
      break;
    } else {
      pre_datalen = datalen;
    }
  }

  tmp = MQTTPacket_read(buf, buflen, getdata);
  if (tmp == CONNACK) {
    unsigned char sessionPresent, connack_rc;
    if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) ==
        1) {
      if (connack_rc == 0) {
        res = 1;
		printf("MqttConnect_Process succeed\r\n");
      } else {
        res = 0;
		printf("MqttConnect_Process failed  0\r\n");
      }
    } else {
	  printf("MqttConnect_Process failed  1\r\n");
      res = 0;
    }
  }else{
     printf("MqttConnect_Process failed  2\r\n");
  }

  process_post(pro, PROCESS_EVENT_CONTINUE, (process_data_t)&res);
  netstatus = net_idle;

  PROCESS_END();
}

PROCESS(MqttSub_Process, "MqttSub");
PROCESS_THREAD(MqttSub_Process, ev, data) {
  static struct etimer et;
  static unsigned long tmp_time;
  static uint8_t datalen = 0;
  static uint8_t pre_datalen = 0;

  PROCESS_BEGIN();

  res = 0;
  netstatus = net_doing;

  if (ev == PROCESS_EVENT_INIT) {
    pro = (struct process *)data;
  }

  sub_Serialize();  //订阅数据序列化

  uart_send((uint8_t *)buf, len);
  memset(buf, 0, buflen);

  pre_datalen = datalen = uart_getlen();
  tmp_time = HAL_GetTick();
  while (1) {
    etimer_set(&et, 5);
    PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
    datalen = uart_getlen();
    if ((datalen == pre_datalen && datalen > 0) ||
        (HAL_GetTick() - tmp_time > 5000)) {
      break;
    } else {
      pre_datalen = datalen;
    }
  }

  tmp = MQTTPacket_read(buf, buflen, getdata);
  if (tmp == SUBACK) {
    unsigned short submsgid;
    int subcount;
    int granted_qos;
    if (MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf,
                               buflen) == 1) {
      res = 1;
	  printf("MqttSub_Process succeed\r\n");
    } else {
      res = 0;
	  printf("MqttSub_Process failed 0\r\n");
    }
  }else{
	printf("MqttSub_Process failed 1\r\n");
  }

  process_post(pro, PROCESS_EVENT_CONTINUE, (process_data_t)&res);
  netstatus = net_idle;

  PROCESS_END();
}

PROCESS(MqttLoop_Process, "MqttLoop");
PROCESS_THREAD(MqttLoop_Process, ev, data) {
  static struct etimer et;
  static uint16_t lose = 0;
  static unsigned long time_up = 0;
  static unsigned long tmp_time;
  static uint16_t updata_time_flag = 0;
  static uint8_t datalen = 0;
  static uint8_t pre_datalen = 0;
  static uint8_t pubqos = PUB_QOS;

  PROCESS_BEGIN();

  time_up = HAL_GetTick();
  lose = 3;
  last_ping_time = 0;
  while (1) {
    netstatus = net_doing;

    if (uart_getlen() >= 2) {
      pre_datalen = datalen = uart_getlen();
      tmp_time = HAL_GetTick();
      while (1) {
        etimer_set(&et, 5);
        PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
        datalen = uart_getlen();
        if ((datalen == pre_datalen && datalen > 0) ||
            (HAL_GetTick() - tmp_time > 100)) {
          break;
        } else {
          pre_datalen = datalen;
        }
      }
      memset(buf, 0, buflen);
      tmp = MQTTPacket_read(buf, buflen, getdata);
      if (tmp > -1) {
      }
      lose = 3;
      if (tmp == PINGRESP) {
      } else if (tmp == PUBLISH) {
        rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid_r, &Topic,
                                     &payload_in, &payloadlen_in, buf, buflen);
        if (rc == 1) {
          if (qos == 0) {
          } else if (qos == 1) {
            static uint8_t len;
            len = MQTTSerialize_puback(buf, buflen, msgid_r);
            uart_send((uint8_t *)buf, len);
            etimer_set(&et, 10);
            PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&et));
          } else {
            static uint8_t len;
            len = MQTTSerialize_pubrec(buf, buflen, 0, msgid_r);
            uart_send((uint8_t *)buf, len);
            etimer_set(&et, 10);
            PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&et));
          }
          analysis_data(payload_in, payloadlen_in);
          etimer_set(&et, 10);
          PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&et));
        }
      } else if (tmp == PUBREL) {
        static uint8_t pack_type = 0;
        rc = MQTTDeserialize_ack(&pack_type, &dup, &msgid_r, buf, buflen);
        if (rc == 1 && pack_type == PUBREL) {
          static uint8_t len;
          len = MQTTSerialize_pubcomp(buf, buflen, msgid_r);
          uart_send((uint8_t *)buf, len);
          etimer_set(&et, 10);
          PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&et));
        }
      } else if (tmp == PUBACK) {
        static uint8_t pack_type = 0;
        rc = MQTTDeserialize_ack(&pack_type, &dup, &msgid_r, buf, buflen);
        if (rc == 1 && pack_type == PUBACK) {
          static uint8_t i = 0;
          for (i = 0; i < mqtt_fifo.Counter; i++) {
            static sMqttData MqttData;
            FIFO_GetOne(&mqtt_fifo, (sMqttData *)&MqttData);
            if (MqttData.msgid == msgid_r) {
              free_mqttdata(MqttData);
            } else {
              FIFO_AddOne(&mqtt_fifo, (sMqttData *)&MqttData);
            }
          }
        }
      }
    }


      if ((HAL_GetTick() - last_ping_time) > 10 * 1000) {
        lose--;
        if (lose == 0) {
		  printf("mqtt ping timeout \r\n");
          if (netdevice_type == net_device_wifi) {
            etimer_set(&et, 1000);
            PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
            uart_send("+++", 3);
//            etimer_set(&et, 20);
//            PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
//            uart_send("+", 1);
//            etimer_set(&et, 20);
//            PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
//            uart_send("+", 1);
            etimer_set(&et, 100);
            PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
            i = 1;
            set_wificmdval(&wificmd, AT_OUTTRANS, AT_OUTTRANS_RES,
                           &MqttLoop_Process);
            wifisend_cmd(WifiCMD_Process, wificmd, i);
            process_start(&WifiInit_Process, NULL);
          } else {
            etimer_set(&et, 3100);
            PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
            Gsm_CIPTMODE_Off();
            etimer_set(&et, 1000);
            PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
            process_start(&GprsInit_Process, NULL);
          }
          process_exit(&MqttLoop_Process);
          netstatus = net_idle;

          return PT_ENDED;
        }
        last_ping_time = HAL_GetTick();
        static uint8_t len;
        len = MQTTSerialize_pingreq(buf, buflen);
        uart_send((uint8_t *)buf, len);
		etimer_set(&et, 100);
        PROCESS_WAIT_EVENT_UNTIL(ev == PROCESS_EVENT_TIMER);
      }
	  
	  if (mqtt_fifo.Counter > 0) {
        static uint8_t msg_len;
        static sMqttData MqttData;
        FIFO_GetOne(&mqtt_fifo, (sMqttData *)&MqttData);
		if(MqttData.dup>1){
			MqttData.dup--;
			FIFO_AddOne(&mqtt_fifo, (sMqttData *)&MqttData);
		}else{//dup为0 1 时才发送
			if(MqttData.dup==1){
				printf("mqtt resend:cmd->%d\r\n",MqttData.cmd);
			}
			msg_len = data_Serialize(MqttData);
			if (msg_len > 0) {
			  len = pub_Serialize(MqttData.eve, databuf, msg_len, MqttData.dup, pubqos,
								  &MqttData.msgid);
			  uart_send((uint8_t *)buf, len);
			  etimer_set(&et, 50);
              PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&et));
			  if (pubqos == 0) {
				free_mqttdata(MqttData);
			  } else {
				MqttData.dup = 250;
				FIFO_AddOne(&mqtt_fifo, (sMqttData *)&MqttData);
			  }
			}
	    }
      }

    if (HAL_GetTick() - time_up > 10 * 1000) {
      time_up = HAL_GetTick();
      updata_time_flag++;
      if (updata_time_flag == 3600) {
        updata_time_flag = 0;
        send_gettime();  //定时更新时间
		printf("updata_time\r\n");
      }
    }

    netstatus = net_idle;
    etimer_set(&et, 10);
    PROCESS_WAIT_EVENT_UNTIL(etimer_expired(&et));
  }
  PROCESS_END();
}

void exit_mqtt(void) {
  process_exit(&MqttInit_Process);
  process_exit(&MqttConnect_Process);
  process_exit(&MqttSub_Process);
  process_exit(&MqttLoop_Process);
}
